/*###########################################################
#   JoinSplitFilterBase
#   for bugs3
#   Wilder, tom
############################################################*/

package biofilter.filters;

import biofilter.exceptions.InvalidRecordException;
import biofilter.records.Record;
import biofilter.sources.RecordsSourceIF;
import java.util.*;

public abstract class JoinSplitFilterBase extends RecordsFilter {
    
    /**
     * Holds the set of records from the source that are needed by the
     * transformRecord method.
     */
    protected LinkedList<Record> inputQueue = new LinkedList<Record>();
    /**
     * Holds the set of transformed records produced by transformRecord().
     * Each time getRecord() is called on this object, the outputQueue is
     * checked for a transformedRecord to return.
     */
    protected LinkedList<Record> outputQueue = new LinkedList<Record>();
    private int requestNum = 0;
    
    public  JoinSplitFilterBase() {
        super();
    }
    
    public  JoinSplitFilterBase(RecordsSourceIF rs) {
        super( rs );
    }
    
    /**
     * Sets the number of records that is retreived from the source before calling
     * transformRecord. The default value is 0, which causes all records from
     * the source to be retreived before transformRecord is called.
     * @param requestNum
     */
    public void setRequestNum(int requestNum) {
        this.requestNum = requestNum;
    }
    
    public Record getRecord() throws InvalidRecordException {
        this.transformedRecord = null;
        if( this.outputQueue.isEmpty()) {
            while ( !eof() && ( requestNum == 0 || inputQueue.size() < requestNum) 
            && (!source.eof())) {
                this.originalRecord = source.getRecord();
                if ( this.originalRecord != null) {
                    if ( trackOriginalRecords) {
                        this.originalRecords.add( this.originalRecord);
                    }
                    this.inputQueue.add( this.originalRecord);
                }
                //else if (source.eof()) {break;}
                // or else we loop forever
            }
            this.transformedRecord = this.transformRecord( null);
            if ( trackTransformedRecords) {
                this.transformedRecords.add( this.transformedRecord);
            }
        }
        else {
            this.transformedRecord = (Record) outputQueue.removeFirst();
            if ( trackTransformedRecords) {
                this.transformedRecords.add( this.transformedRecord);
            }
        }
        return this.transformedRecord;
    }
    
    public boolean eof() {
        
        return ( this.source.eof() && this.outputQueue.isEmpty()
        && ( this.requestNum == 0 || inputQueue.isEmpty()));
    }
    
}
